package com.atguigu.chapter12;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @ClassName: Flink01_Project_TopN
 * @Description:
 * @Author: kele
 * @Date: 2021/4/15 9:41
 *
 * 12.1	使用SQL实现热门商品TOP N
 *
 * 每隔10min 统计最近 1hour的热门商品 top3, 并把统计的结果写入到mysql中
 *
 * 1、获取数据
 * 2、group窗口，将同一窗口下同一商品进行分组、获取商品的访问量
 * 3、通过over对窗口的数据进行排序，
 * 4、获取前三的商品
 * 5、将数据写入到mysql
 *      --注意字段、类型要一致，否则无法写入
 *
 **/
public class Flink01_Project_TopN {

    public static void main(String[] args) {

        Configuration conf = new Configuration();
        conf.setInteger("rest.port",20000);

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);

        env.setParallelism(2);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        /**
         * 获取数据
         */
         tenv.executeSql("create table user_behavior(\n" +
                "user_id bigint, \n" +
                "item_id bigint, \n" +
                "category_id int,\n" +
                "behavior string,\n" +
                "ts bigint,\n" +
                "ets as to_timestamp(from_unixtime(ts,'yyyy-MM-dd HH:mm:ss')),\n" +
                "watermark for ets as ets - interval '5' second)\n" +
                "with(\n" +
                "'connector' = 'filesystem',\n" +
                "'path' = 'in/UserBehavior.csv',\n" +
                "'format' = 'csv'\n" +
                ")");

        Table table = tenv.sqlQuery("select item_id,\n" +
                "count(*) num,\n" +
                "HOP_END(ets,interval '10' minute,interval '1' hour) wend\n" +
                "from user_behavior\n" +
                "where behavior='pv' \n" +
                "group by item_id,HOP(ets,interval '10' minute,interval '1' hour)");

        tenv.createTemporaryView("t1",table);

        /**
         * 对在同一个窗口的所有的数据进行 排序
         */
        Table table2 = tenv.sqlQuery("select item_id,wend,\n" +
                "num,row_number() over(partition by wend order by num desc) rk\n" +
                "from t1\n");

        tenv.createTemporaryView("t2",table2);

        Table table3 = tenv.sqlQuery("select item_id,wend,\n" +
                "num,rk\n" +
                "from t2\n" +
                "where rk<=3");
/**
 * 注册一个mysql的表
 */

tenv.executeSql("CREATE TABLE hot_item (\n" +
        "  item_id BIGINT,\n" +
        "  wend timestamp(3),\n" +
        "  num BIGINT NOT NULL,\n" +
        "  rk BIGINT NOT NULL,\n" +
        "  PRIMARY KEY (wend,rk) NOT ENFORCED\n" +
        ") WITH (\n" +
        "   'connector' = 'jdbc',\n" +
        "   'url' = 'jdbc:mysql://hadoop162:3306/test',\n" +
        "   'table-name' = 'hot_item',\n" +
        "   'username' = 'root',\n" +
        "   'password' = 'aaaaaa'\n" +
        ")");


        table3.executeInsert("hot_item");

    }

}
